package xubo.wangcaifeng.love

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.spark.rdd.RDD


object Utils {

  //计算总的订单量
  def orderSum(lines: RDD[String]) = {

    val value: RDD[(String ,Int)] = lines.map(line => {
      val json: JSONObject = JSON.parseObject(line)
      val serviceName: String = json.get("serviceName").toString
      val everyday: String = json.get("requestId").toString.take(8)
      if (serviceName.equals("reChargeNotifyReq")) {
        ("OrderSum", 1)
      } else {
        ("OrderSum", 0)
      }
    })
    //存到redis
    value.foreachPartition({
          //还在dirver端
      line =>
        val jedis = JedisConnectionPool.getConnection()
        line.foreach(x => {
          jedis.incrBy(x._1, x._2)
        })
        jedis.close()
    })
  }
  //计算总的金额
  def SumMoney(lines: RDD[String]) = {
    val chargefeeSum: RDD[(String, Long)] = lines.map(line => {
      val json = JSON.parseObject(line)
      val boolean1 = json.get("serviceName").toString.equals("reChargeNotifyReq")
      val boolean2 = json.get("bussinessRst").toString.equals("0000")
      val chargefee = json.get("chargefee").toString.toLong
      if (boolean1 && boolean2) {
        ("chargefeeSum", chargefee)
      } else {
        ("chargefeeSum", 0)
      }
    })
    //存入到redis
    chargefeeSum.foreachPartition(line=>{
      val jedis = JedisConnectionPool.getConnection()
      line.foreach(x=>
          jedis.incrBy(x._1,x._2)
      )
      jedis.close()
    })
  }
  //计算充值成功率
  def SuccessRate(lines: RDD[String]) = {
    val success: RDD[(String, Int)] = lines.map(line => {
      val json: JSONObject = JSON.parseObject(line)
      val serviceName: String = json.get("serviceName").toString
      val success = json.get("bussinessRst").toString

      if (serviceName.equals("reChargeNotifyReq") && success.equals("0000")) {
        ("SuccessSum", 1)
      } else {
        ("SuccessSum", 0)
      }
    })
    //存入到redis
    success.foreachPartition(line=>{
      val jedis = JedisConnectionPool.getConnection()
      line.foreach(x=>
        jedis.incrBy(x._1,x._2)
      )
      jedis.close()
    })
    //计算成功率
    val jedis = JedisConnectionPool.getConnection()
    val double: Double = jedis.get("SuccessSum").toDouble
    val double1: Double = jedis.get("OrderSum").toDouble
    /*var l = double/double1
    println(double1+double+l)*/
    var SuccessRate = double/double1
    jedis.incrByFloat("SuccessRate",SuccessRate)
    jedis.close()
  }
  //计算平均时长
  def avgTime(lines: RDD[String]) = {
    val DifferentTime: RDD[(String, Double)] = lines.map(line => {
      val json: JSONObject = JSON.parseObject(line)
      val serviceName: String = json.get("serviceName").toString
      if (serviceName.equals("reChargeNotifyReq")){
        var startTime = json.get("requestId").toString.toDouble
        var endTime = json.get("receiveNotifyTime").toString.substring(0, 17).toDouble
        var differentTime = endTime - startTime
        ("differentTime", differentTime)
      }else {
        ("differentTime",0)
      }
    })
    //存入redis
    DifferentTime.foreachPartition(line=>{
      val jedis = JedisConnectionPool.getConnection()
      line.foreach(x=>
        jedis.incrByFloat(x._1,x._2)
      )
      jedis.close()
    })
    //计算平均时长
    val jedis = JedisConnectionPool.getConnection()
    var OrderSum: Double = jedis.get("OrderSum").toDouble
    var SumTime: Double = jedis.get("differentTime").toDouble
    var SumSuccessRate: Double = OrderSum/SumTime
    jedis.set("SumSuccessRate",SumSuccessRate.formatted("%.2f"))
    jedis.close()
  }
  //计算总共的订单量
  def topOrder(lines: RDD[String]) = {
    val Toporder: RDD[(String, Int)] = lines.map(line => {
      val json: JSONObject = JSON.parseObject(line)
      if (json.get("serviceName")!=null) {
        ("TopOrder", 1)
      } else {
        ("TopOrder", 0)
      }
    })
    //存到redis
    Toporder.foreachPartition(
      line=>{
        val jedis = JedisConnectionPool.getConnection()
        line.foreach(x=>{
          jedis.incrByFloat(x._1,x._2)
        })
        jedis.close()
      }
    )
  }
  //计算总共的成功量
  def topOrderSuccess(lines: RDD[String]) = {
    val TopSuccess: RDD[(String, Int)] = lines.map(line => {
      val json: JSONObject = JSON.parseObject(line)
      if (json.get("serviceName") != null && json.get("bussinessRst").toString.equals("0000")) {
        ("TopOrderSuccess", 1)
      } else {
        ("TopOrderSuccess", 0)
      }
    })
    //存到redis
    TopSuccess.foreachPartition(
      line=>{
        val jedis = JedisConnectionPool.getConnection()
        line.foreach(x=>{
          jedis.incrByFloat(x._1,x._2)
        })
        jedis.close()
      }
    )
    val jedis = JedisConnectionPool.getConnection()
    val double = jedis.get("TopOrderSuccess").toDouble
    val double1 = jedis.get("TopOrder").toDouble
    println(double)
    println(double1)
    var TopSuccessRate = double/double1
    jedis.set("TopSuccessRate",TopSuccessRate.formatted("%.2f"))
  }

}
